package io.debezium.connector.postgresql;

import io.debezium.annotation.ThreadSafe;
import io.debezium.connector.common.CdcSourceTaskContext;
import io.debezium.connector.postgresql.connection.GaussDBConnection;
import io.debezium.connector.postgresql.connection.ReplicationConnection;
import io.debezium.connector.postgresql.spi.SlotState;
import io.debezium.relational.TableId;
import io.debezium.schema.TopicSelector;
import io.debezium.util.Clock;
import io.debezium.util.ElapsedTimeStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.util.Collections;

@ThreadSafe
public class PostgresTaskContext extends CdcSourceTaskContext {
    protected static final Logger LOGGER = LoggerFactory.getLogger(PostgresTaskContext.class);
    private final PostgresConnectorConfig config;
    private final TopicSelector<TableId> topicSelector;
    private final PostgresSchema schema;
    private ElapsedTimeStrategy refreshXmin;
    private Long lastXmin;

    protected PostgresTaskContext(
            PostgresConnectorConfig config,
            PostgresSchema schema,
            TopicSelector<TableId> topicSelector) {
        super(config.getContextName(), config.getLogicalName(), Collections::emptySet);
        this.config = config;
        if (config.xminFetchInterval().toMillis() > 0L) {
            this.refreshXmin =
                    ElapsedTimeStrategy.constant(
                            Clock.SYSTEM, config.xminFetchInterval().toMillis());
        }

        this.topicSelector = topicSelector;

        assert schema != null;

        this.schema = schema;
    }

    protected TopicSelector<TableId> topicSelector() {
        return this.topicSelector;
    }

    protected PostgresSchema schema() {
        return this.schema;
    }

    protected PostgresConnectorConfig config() {
        return this.config;
    }

    protected void refreshSchema(GaussDBConnection connection, boolean printReplicaIdentityInfo)
            throws SQLException {
        this.schema.refresh(connection, printReplicaIdentityInfo);
    }

    Long getSlotXmin(GaussDBConnection connection) throws SQLException {
        if (this.config.xminFetchInterval().toMillis() <= 0L) {
            return null;
        } else {
            assert this.refreshXmin != null;

            if (this.refreshXmin.hasElapsed()) {
                this.lastXmin = this.getCurrentSlotState(connection).slotCatalogXmin();
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Fetched new xmin from slot of {}", this.lastXmin);
                }
            } else if (LOGGER.isTraceEnabled()) {
                LOGGER.trace("reusing xmin value of {}", this.lastXmin);
            }

            return this.lastXmin;
        }
    }

    private SlotState getCurrentSlotState(GaussDBConnection connection) throws SQLException {
        return connection.getReplicationSlotState(
                this.config.slotName(), this.config.plugin().getPostgresPluginName());
    }

    protected ReplicationConnection createReplicationConnection(
            boolean doSnapshot, GaussDBConnection jdbcConnection) throws SQLException {
        boolean dropSlotOnStop = this.config.dropSlotOnStop();
        if (dropSlotOnStop) {
            LOGGER.warn(
                    "Connector has enabled automated replication slot removal upon restart ({} = true). This setting is not recommended for production environments, as a new replication slot will be created after a connector restart, resulting in missed data change events.",
                    PostgresConnectorConfig.DROP_SLOT_ON_STOP.name());
        }

        return ReplicationConnection.builder(this.config)
                .withSlot(this.config.slotName())
                .withPublication(this.config.publicationName())
                .withTableFilter(this.config.getTableFilters())
                .withPublicationAutocreateMode(this.config.publicationAutocreateMode())
                .withPlugin(this.config.plugin())
                .dropSlotOnClose(dropSlotOnStop)
                .streamParams(this.config.streamParams())
                .statusUpdateInterval(this.config.statusUpdateInterval())
                .withTypeRegistry(this.schema.getTypeRegistry())
                .doSnapshot(doSnapshot)
                .withSchema(this.schema)
                .jdbcMetadataConnection(jdbcConnection)
                .build();
    }

    PostgresConnectorConfig getConfig() {
        return this.config;
    }
}
